package com.example.rxjava;


import org.reactivestreams.Subscriber;

import java.sql.Time;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.CompletableObserver;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableEmitter;
import io.reactivex.rxjava3.core.ObservableOnSubscribe;
import io.reactivex.rxjava3.core.ObservableSource;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.core.SingleObserver;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.Action;
import io.reactivex.rxjava3.functions.BiFunction;
import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.functions.Function;
import io.reactivex.rxjava3.functions.Function3;
import io.reactivex.rxjava3.functions.Predicate;
import io.reactivex.rxjava3.observables.GroupedObservable;
import io.reactivex.rxjava3.schedulers.Schedulers;

public class MainActivity  {
    public static void main(String[] args) {
//        Observable.create(new ObservableOnSubscribe<String>() {
//            @Override
//            public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Throwable {
//              emitter.onNext("杨影月");
//              emitter.onNext("月眉儿");
//              emitter.onComplete();
//            }
//        });
//        String[] strings = {"杨影月","月眉儿"};
//        Observable.fromArray(strings);
//        Observable.just("杨影月","月眉儿").subscribe(new Observer<String>() {
//            @Override
//            public void onSubscribe(@NonNull Disposable d) {
//                System.out.println("onSubscribe"+d.toString());
//            }
//
//            @Override
//            public void onNext(@NonNull String s) {
//                System.out.println("onNext"+s);
//            }
//
//            @Override
//            public void onError(@NonNull Throwable e) {
//                System.out.println("onError"+e.getLocalizedMessage());
//            }
//
//            @Override
//            public void onComplete() {
//                System.out.println("onComplete");
//            }
//        });
//        Consumer<String> onNext = new Consumer<String>() {
//            @Override
//            public void accept(String s) throws Throwable {
//                System.out.println("onNext.."+s);
//            }
//        };
//        Consumer<Throwable> onError = new Consumer<Throwable>() {
//            @Override
//            public void accept(Throwable throwable) throws Throwable {
//                System.out.println("onError.."+throwable.getLocalizedMessage());
//            }
//        };
//        Action onComplete = new Action() {
//            @Override
//            public void run() throws Throwable {
//               System.out.println("Complete..完成");
//            }
//        };
//        Observable.just("99","66").subscribe(onNext,onError,onComplete);
//        Observable.interval(3, TimeUnit.SECONDS)
//                .subscribe(new Consumer<Long>() {
//            @Override
//            public void accept(Long aLong) throws Throwable {
//                System.out.println(aLong+"");
//            }
//        });
//        Observable.range(1,5).repeat(2).subscribe(new Consumer<Integer>() {
//            @Override
//            public void accept(Integer integer) throws Throwable {
//                System.out.println(integer+"");
//            }
//        });
        final String Host = "http://blog.csdn.net/";
//        Observable.just("it").map(new Function<String, String>() {
//            @Override
//            public String apply(String s) throws Throwable {
//                return Host+s;
//            }
//        }).subscribe(new Consumer<String>() {
//            @Override
//            public void accept(String s) throws Throwable {
//                System.out.println(s);
//            }
//        });
//        Observable.just("it1","58y","fry").concatMap(new Function<String, ObservableSource<?>>() {
//            @Override
//            public ObservableSource<?> apply(String s) throws Throwable {
//                return Observable.just(Host+s);
//            }
//        }).cast(String.class).subscribe(new Consumer<String>() {
//            @Override
//            public void accept(String s) throws Throwable {
//                System.out.println(s);
//            }
//        });
//        Observable.just(1,2,3).flatMapIterable(new Function<Integer, Iterable<Integer>>() {
//            @Override
//            public Iterable<Integer> apply(Integer integer) throws Throwable {
//                List<Integer> integers = new ArrayList<>();
//                integers.add(integer+1);
//                return integers;
//            }
//        }).subscribe(new Consumer<Integer>() {
//            @Override
//            public void accept(Integer integer) throws Throwable {
//                System.out.println(""+integer);
//            }
//        });
//        Observable.just(1,2,3,4,5,6,7).buffer(3).subscribe(new Consumer<List<Integer>>() {
//            @Override
//            public void accept(List<Integer> integers) throws Throwable {
//               for (Integer i :integers){
//                   System.out.println("buffer"+i);
//               }
//                System.out.println("----------");
//            }
//        });
//        Swordsman s1 = new Swordsman("123", "A");
//        Swordsman s2 = new Swordsman("re", "SS");
//        Swordsman s3 = new Swordsman("er", "S");
//        Swordsman s4 = new Swordsman("dg", "A");
//        Swordsman s5 = new Swordsman("hg", "AA");
//        Swordsman s6 = new Swordsman("cb", "SS");
//        Swordsman s7 = new Swordsman("mj", "AA");
//        Swordsman s8 = new Swordsman("tgb", "A");
//        Observable<GroupedObservable<String, Swordsman>> groupedObservableObservable = Observable.just(s1, s2, s3, s4, s5, s6, s7, s8)
//                .groupBy(new Function<Swordsman, String>() {
//            @Override
//            public String apply(Swordsman swordsman) throws Throwable {
//                return swordsman.getA();
//            }
//        });
//        Observable.concat(groupedObservableObservable).subscribe(new Consumer<Swordsman>() {
//            @Override
//            public void accept(Swordsman swordsman) throws Throwable {
//                System.out.println(swordsman.getS()+"---"+swordsman.getA());
//            }
//        });
//        try {
//            Thread.sleep(3000);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
//        Observable.just(1,2,3,4,5,6).filter(new Predicate<Integer>() {
//            @Override
//            public boolean test(Integer integer) throws Throwable {
//                return integer>3;
//            }
//        }).subscribe(new Consumer<Integer>() {
//            @Override
//            public void accept(Integer integer) throws Throwable {
//                System.out.println("filter"+integer);
//            }
//        });
//        Observable.just(1,2,3,4,5,6).elementAt(2).subscribe(new Consumer<Integer>() {
//            @Override
//            public void accept(Integer integer) throws Throwable {
//                System.out.println("elementAt"+integer);
//            }
//        });
//        Observable.just(1,2,3,3,4,5,3,6).distinct().subscribe(new Consumer<Integer>() {
//            @Override
//            public void accept(Integer integer) throws Throwable {
//                System.out.println("distinct"+integer);
//            }
//        });
//        Observable.just(1,2,3,4,5).skip(2).subscribe(new Consumer<Integer>() {
//            @Override
//            public void accept(Integer integer) throws Throwable {
//                System.out.println("skip"+integer);
//            }
//        });
//        Observable.just(1,2,3,4,5).skipLast(2).subscribe(new Consumer<Integer>() {
//            @Override
//            public void accept(Integer integer) throws Throwable {
//                System.out.println("take"+integer);
//            }
//        });
//        Observable.just(1,2,3,4,5,6).ignoreElements().subscribe(new CompletableObserver() {
//            @Override
//            public void onSubscribe(@NonNull Disposable d) {
//
//            }
//
//            @Override
//            public void onComplete() {
//                System.out.println("完成");
//            }
//
//            @Override
//            public void onError(@NonNull Throwable e) {
//
//            }
//        });
//        Observable.create(new ObservableOnSubscribe<Integer>() {
//            @Override
//            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Throwable {
//                for (int i = 0;i<10;i++){
//                    emitter.onNext(i);
//                    Thread.sleep(100);
//                }
//                emitter.onComplete();
//            }
//        }).throttleFirst(200,TimeUnit.MILLISECONDS).subscribe(new Consumer<Integer>() {
//            @Override
//            public void accept(Integer integer) throws Throwable {
//                System.out.println("throttleFirst"+integer);
//            }
//        });
//        Observable.create(new ObservableOnSubscribe<Integer>() {
//            @Override
//            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Throwable {
//                for (int i = 0;i<10;i++){
//                    emitter.onNext(i);
//                    int sleep = 100;
//                    if(i%3==0){
//                        sleep = 300;
//                    }
//                    Thread.sleep(sleep);
//                }
//                emitter.onComplete();
//            }
//        }).throttleWithTimeout(200,TimeUnit.MILLISECONDS).subscribe(new Consumer<Integer>() {
//            @Override
//            public void accept(Integer integer) throws Throwable {
//                System.out.println(""+integer);
//            }
//        });
//        Observable.just(3,4,5).startWith(Observable.fromArray(1,2)).subscribe(new Consumer<Integer>() {
//            @Override
//            public void accept(Integer integer) throws Throwable {
//                System.out.println(""+integer);
//            }
//        });
        Observable<Integer> obs1 = Observable.just(3, 4, 5).subscribeOn(Schedulers.io());
        Observable<String> obs2 = Observable.just("a", "b", "c");
//        Observable.merge(obs2,obs1).subscribe(new Consumer<Integer>() {
//            @Override
//            public void accept(Integer integer) throws Throwable {
//                System.out.println(""+integer);
//            }
//        });
//        Observable.concat(obs2,obs1).subscribe(new Consumer<Integer>() {
//            @Override
//            public void accept(Integer integer) throws Throwable {
//                System.out.println(""+integer);
//            }
//        });
//        Observable.zip(obs1, obs2, new BiFunction<Integer, String, String>() {
//            @Override
//            public String apply(Integer integer, String s) throws Throwable {
//                return integer+s;
//            }
//        }).subscribe(new Consumer<String>() {
//            @Override
//            public void accept(String s) throws Throwable {
//                System.out.println(s);
//            }
//        });
//        Observable.combineLatest(obs1, obs2, new BiFunction<Integer, String, String>() {
//            @Override
//            public String apply(Integer integer, String s) throws Throwable {
//                return integer+s;
//            }
//        }).subscribe(new Consumer<String>() {
//            @Override
//            public void accept(String s) throws Throwable {
//                System.out.println(s);
//            }
//        });
//        Observable.create(new ObservableOnSubscribe<Long>(){
//
//            @Override
//            public void subscribe(@NonNull ObservableEmitter<Long> emitter) throws Throwable {
//                long l = System.currentTimeMillis() / 1000;
//                emitter.onNext(l);
//            }
//        }).delay(2,TimeUnit.SECONDS).subscribe(new Consumer<Long>() {
//            @Override
//            public void accept(Long aLong) throws Throwable {
//                System.out.println(System.currentTimeMillis() / 1000-aLong);
//            }
//        });
//        Observable.just(1,2).doOnNext(new Consumer<Integer>() {
//            @Override
//            public void accept(Integer integer) throws Throwable {
//                System.out.println(integer+"||||");
//            }
//        }).subscribe(new Observer<Integer>() {
//            @Override
//            public void onSubscribe(@NonNull Disposable d) {
//
//            }
//
//            @Override
//            public void onNext(@NonNull Integer integer) {
//                System.out.println(integer+"&&&");
//            }
//
//            @Override
//            public void onError(@NonNull Throwable e) {
//
//            }
//
//            @Override
//            public void onComplete() {
//
//            }
//        });
//        Observable<String> obs = Observable.create(new ObservableOnSubscribe<String>() {
//            @Override
//            public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Throwable {
//                String name = Thread.currentThread().getName();
//                emitter.onNext(name);
//                System.out.println(Thread.currentThread().getName()+"345");
//                emitter.onComplete();
//            }
//        });
//        obs.subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<String>() {
//            @Override
//            public void accept(String s) throws Throwable {
//                System.out.println(Thread.currentThread().getName()+"345");
//            }
//        });
//        try {
//            Thread.sleep(3000);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
//        Observable.create(new ObservableOnSubscribe<Integer>() {
//            @Override
//            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Throwable {
//                for (int i=0;i<4;i++) {
//                    Thread.sleep(200*i);
//                    emitter.onNext(i);
//                }
//                emitter.onComplete();
//            }
//        }).timeout(200,TimeUnit.MILLISECONDS,Observable.just(22,33)).subscribe(new Consumer<Integer>() {
//            @Override
//            public void accept(Integer integer) throws Throwable {
//                System.out.println("s"+integer);
//            }
//        });
//        Observable.create(new ObservableOnSubscribe<Integer>() {
//            @Override
//            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Throwable {
//                for (int i=0;i<6;i++) {
//                    if(i>2){
//                        emitter.onError(new Throwable("Throwable Error"));
//                    }
//                    emitter.onNext(i);
//                }
//                emitter.onComplete();
//            }
//        }).onErrorReturn(new Function<Throwable, Integer>() {
//            @Override
//            public Integer apply(Throwable throwable) throws Throwable {
//                return 8;
//            }
//        }).subscribe(new Observer<Integer>() {
//            @Override
//            public void onSubscribe(@NonNull Disposable d) {
//                System.out.println("onSubscribe.."+d.toString());
//            }
//
//            @Override
//            public void onNext(@NonNull Integer integer) {
//                System.out.println("onNext.."+integer);
//            }
//
//            @Override
//            public void onError(@NonNull Throwable e) {
//                System.out.println("onError.."+e.getLocalizedMessage());
//            }
//
//            @Override
//            public void onComplete() {
//                System.out.println("onComplete..");
//            }
//        });
//        Observable.create(new ObservableOnSubscribe<Integer>() {
//            @Override
//            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Throwable {
//                for (int i=0;i<5;i++) {
//                    if(i==1){
//                        emitter.onError(new Throwable("Throwable Error"));
//                    }else{
//                        emitter.onNext(i);
//                    }
//                }
//                emitter.onComplete();
//            }
//        }).retry(2).subscribe(new Observer<Integer>() {
//            @Override
//            public void onSubscribe(@NonNull Disposable d) {
//                System.out.println("onSubscribe.."+d.toString());
//            }
//
//            @Override
//            public void onNext(@NonNull Integer integer) {
//                System.out.println("onNext.."+integer);
//            }
//
//            @Override
//            public void onError(@NonNull Throwable e) {
//                System.out.println("onError.."+e.getLocalizedMessage());
//            }
//
//            @Override
//            public void onComplete() {
//                System.out.println("onComplete..");
//            }
//        });
//        Observable.just(1,2,3,4).all(new Predicate<Integer>() {
//            @Override
//            public boolean test(Integer integer) throws Throwable {
//                return integer<3;
//            }
//        }).subscribe(new SingleObserver<Boolean>() {
//            @Override
//            public void onSubscribe(@NonNull Disposable d) {
//                System.out.println("onSubscribe"+d.toString());
//            }
//
//            @Override
//            public void onSuccess(@NonNull Boolean aBoolean) {
//                System.out.println("onSuccess"+aBoolean);
//            }
//
//            @Override
//            public void onError(@NonNull Throwable e) {
//                System.out.println("onSuccess"+e.getLocalizedMessage());
//            }
//        });
//        Observable.just(1,2,3,4).contains(1).subscribe(new Consumer<Boolean>() {
//            @Override
//            public void accept(Boolean aBoolean) throws Throwable {
//                System.out.println("accept.."+aBoolean);
//            }
//        });
//        Observable.create(new ObservableOnSubscribe<Integer>() {
//            @Override
//            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Throwable {
//                emitter.onComplete();
//            }
//        }).isEmpty().subscribe(new Consumer<Boolean>() {
//            @Override
//            public void accept(Boolean aBoolean) throws Throwable {
//                System.out.println("accept..."+aBoolean);
//            }
//        });
//        Observable.ambArray(Observable.just(1,2,3).delay(2,TimeUnit.SECONDS),Observable.just(4,5,6))
//                .subscribe(new Consumer<Integer>() {
//                    @Override
//                    public void accept(Integer integer) throws Throwable {
//                        System.out.println("accept"+integer);
//                    }
//                });
//        Observable.create(new ObservableOnSubscribe<Integer>() {
//            @Override
//            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Throwable {
//                emitter.onComplete();
//            }
//        }).defaultIfEmpty(3).subscribe(new Consumer<Integer>() {
//            @Override
//            public void accept(Integer integer) throws Throwable {
//                System.out.println("accept"+integer);
//            }
//        });
        Observable.just(1,4,3).toSortedList().subscribe(new Consumer<List<Integer>>() {
            @Override
            public void accept(List<Integer> integers) throws Throwable {
                for (int i:integers) {
                    System.out.println("accept" + i);
                }
            }
        });
    }
}